import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

public class Firstn {

    public static void main(String[] args) throws Exception {


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String rootPath = new StringBuilder(System.getProperty("user.dir")).append("/dataset_api/Java").toString();

        DataSet<Tuple2<String,Integer>> in =env.readCsvFile("file:///"+rootPath+"/"+"firstn.csv").types(String.class, Integer.class);


        // Return the first five (arbitrary) elements of the DataSet
        DataSet<Tuple2<String, Integer>> out1 = in.first(5);

        System.out.print("---------------------out1----------直接返回前面5条数据------------------------\n");
        out1.print();

        // Return the first two (arbitrary) elements of each String group
        DataSet<Tuple2<String, Integer>> out2 = in.groupBy(0)
                .first(2);

        System.out.print("---------------------out2---------按照key0归类,每个类别返回前面2条数据-------------------------\n");
        out2.print();

        // Return the first three elements of each String group ordered by the Integer field
        DataSet<Tuple2<String, Integer>> out3 = in.groupBy(0)
                .sortGroup(1, Order.ASCENDING)
                .first(3);

        System.out.print("---------------------out3--------------按照key0归类,每个类别返回前面3条数据--------------------\n");
        out3.print();

    }

}
